package rocket;

import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;
import com.alibaba.rocketmq.client.consumer.PullResult;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.message.MessageQueue;

import java.util.List;
import java.util.Set;

/**
 * 消费者采取拉模式
 * @author wsz
 * @date 2022/2/22 18:14
 **/
public class PullConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("consumer_grp_wsz");
        consumer.setNamesrvAddr("192.168.174.138:9876");
        consumer.start();

        final Set<MessageQueue> messageQueues = consumer.fetchSubscribeMessageQueues("tp_demo1");
        for (MessageQueue messageQueue: messageQueues) {
            /**
             * 消息对象
             * tag进行过滤
             * 偏移量，消费起步的位置
             * 每次最多拉取的消息条数
             */
            PullResult result = consumer.pull(messageQueue, "*", 0, 10);

            final List<MessageExt> msgFoundList = result.getMsgFoundList();
            if (msgFoundList == null) continue;
            for (MessageExt messageExt : msgFoundList) {
                System.out.println(messageExt);
                System.out.println(new String(messageExt.getBody(), "utf8"));
            }
        }

        Thread.sleep(10000);

        consumer.shutdown();
    }
}
